package com.wdciti.sccp.iot.position.service;

import com.wdciti.sccp.iot.position.model.MeasureDataSet;
import com.wdciti.sccp.iot.position.model.MeasureDataValue;
import com.wdciti.sccp.iot.tsdata.core.api.TSDBWrapper;
import com.wdciti.sccp.iot.tsdata.core.api.TsDataException;
import com.wdciti.sccp.iot.tsdata.core.model.DataType;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.util.*;

@Slf4j
@Service
public class PositionDataServiceImpl implements PositionDataService{
    private final TSDBWrapper<SessionPool> tsdbWrapper;

    @PreDestroy
    public void preDestroy() {
        if(tsdbWrapper.getSessionWrapper()!= null){
            this.tsdbWrapper.close();
        }
    }
    public PositionDataServiceImpl(TSDBWrapper<SessionPool> tsdbWrapper) {
        this.tsdbWrapper = tsdbWrapper;
    }

    private SessionPool sessionWrapper() {
        return this.tsdbWrapper.getSessionWrapper();
    }

    @Override
    public List<MeasureDataValue> getLastData(@NonNull String path) {
        String lastQueryFormat = "select last* from %s";
        String sql = String.format(lastQueryFormat, path);
        try (SessionDataSetWrapper dataSet = sessionWrapper().executeQueryStatement(sql)) {
            return buildLastResult(dataSet);
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            log.error(e.getMessage());
            throw new TsDataException(e);
        }
    }

    @Override
    public List<MeasureDataValue> getLastData(@NonNull String path, long timeStamp, long timeout) {

        String queryByTimeFormat = "select last* from %s where timestamp<=%s";
        String sql = String.format(queryByTimeFormat, path, timeStamp);
        try (SessionDataSetWrapper dataSet = sessionWrapper().executeQueryStatement(sql,timeout)) {
            return buildLastResult(dataSet);
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            log.error(e.getMessage());
            throw new TsDataException(e);
        }
    }

    @Override
    public List<MeasureDataValue> getLastData(String path, List<String> keys) {
        if (keys.isEmpty()) {
            return getLastData(path);
        }
        String[] ss = keys.stream().map(s -> path + "." + s).toArray(String[]::new);
        try (SessionDataSetWrapper wrapper = sessionWrapper().executeLastDataQuery(Arrays.asList(ss))) {
            return buildLastResult(wrapper);
        } catch (StatementExecutionException | IoTDBConnectionException e) {
            log.error(e.getMessage());
            return null;
        }
    }

    @Override
    public List<MeasureDataValue> getLastData(String path, List<String> keys, long timestamp, long timeout) {
        if (keys.isEmpty()) {
            return getLastData(path, timestamp, timeout);
        }
        String[] ss = keys.stream().map(s -> path + "." + s).toArray(String[]::new);
        try (SessionDataSetWrapper wrapper = sessionWrapper().executeLastDataQuery(Arrays.asList(ss), timestamp, timeout)) {
            return buildLastResult(wrapper);
        } catch (StatementExecutionException | IoTDBConnectionException e) {
            log.error(e.getMessage());
            throw new TsDataException(e);
        }
    }

    @Override
    public Map<String, List<MeasureDataValue>> getLastDataFromPaths(List<String> paths, List<String> keys) {
        return getLastDataFromPaths(paths, keys, System.currentTimeMillis(),1000);
    }

    @Override
    public Map<String, List<MeasureDataValue>> getLastDataFromPaths(List<String> paths, List<String> keys, long timestamp, long timeout) {
        Map<String, List<MeasureDataValue>> map = new HashMap<>();
        for (String path : paths) {
            try {
                List<MeasureDataValue> data = getLastData(path, keys, timestamp, timeout / paths.size());
            } catch (TsDataException e) {
                log.warn("getLastDataFromPaths() throw TsDataException:" + e.getMessage(), e);
                map.put(path, null);
            }
            map.put(path, null);
        }
        return map;
    }

    @Override
    public List<MeasureDataValue> getDataAtTime(@NonNull String path, long timeStamp) {
        return getDataAtTime(path,null,timeStamp);
    }

    @Override
    public List<MeasureDataValue> getDataAtTime(@NonNull String path, List<String> keys, long timeStamp) {
        String keysString = "*";
        if (keys != null) {
            keysString = Arrays.toString(keys.toArray())
                    .replace("[", "").replace("]", "");
        }
        String queryByTimeFormat = "select last %s from %s where timestamp=%s";
        String sql = String.format(queryByTimeFormat, keysString, path, timeStamp);
        try (SessionDataSetWrapper dataSet = sessionWrapper().executeQueryStatement(sql)) {
            return buildLastResult(dataSet);
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            log.error(e.getMessage());
            throw new TsDataException(e);
        }
    }

    @Override
    public MeasureDataSet getRawData(@NonNull String path, @NonNull List<String> keys, long startTime, long endTime, int limit, int offset) {
        String keysString = String.join(",", keys);
        String sql_format = "select %s from %s where timestamp >= %s and timestamp<=%s limit %s offset %s";
        String sql = String.format(sql_format, keysString, path, startTime, endTime, limit, offset);
        try (SessionDataSetWrapper wrapper = sessionWrapper().executeQueryStatement(sql)) {
            return buildResult(wrapper, new MeasureDataSet(keys));
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            log.error(e.getMessage());
            throw new TsDataException(e);
        }
    }

    @Override
    public MeasureDataSet getRawData(@NonNull String path, long startTime, long endTime, int limit, int offset) {
        String sql_format = "select * from %s where timestamp >= %s and timestamp<=%s limit %s offset %s";
        String sql = String.format(sql_format, path, startTime, endTime, limit, offset);
        try (SessionDataSetWrapper wrapper = sessionWrapper().executeQueryStatement(sql)) {
            return buildResult(wrapper, null);
        } catch (IoTDBConnectionException | StatementExecutionException e) {
            log.error(e.getMessage());
            throw new TsDataException(e);
        }
    }

    @Override
    public MeasureDataSet getRawData(String path, long startTime, long endTime) {
        return PositionDataService.super.getRawData(path, startTime, endTime);
    }

    private List<MeasureDataValue> buildLastResult(SessionDataSetWrapper wrapper) throws IoTDBConnectionException, StatementExecutionException {
        List<MeasureDataValue> result = new ArrayList<>();
        while (wrapper.hasNext()) {
            RowRecord record = wrapper.next();
            if (record != null) {
                List<Field> fieldList = record.getFields();
                String seriesPath = fieldList.get(wrapper.getColumnNames().indexOf("Timeseries") - 1).getStringValue();
                String[] ss = seriesPath.split("\\.");
                String key = ss[ss.length - 1];
                String value = fieldList.get(wrapper.getColumnNames().indexOf("Value") - 1).getStringValue();
                String datatype = fieldList.get(wrapper.getColumnNames().indexOf("DataType") - 1).getStringValue();
                MeasureDataValue dataValue = new MeasureDataValue(record.getTimestamp(), key);
                dataValue.setDataType(DataType.valueOf(datatype));
                dataValue.setStrValue(value);
                result.add(dataValue);
            }
        }
        return result;
    }

    private MeasureDataSet buildResult(SessionDataSetWrapper wrapper, MeasureDataSet dataSet) throws IoTDBConnectionException, StatementExecutionException {
        List<String> columnList = wrapper.getColumnNames();
        if (dataSet == null) {
            List<String> keys = new ArrayList<>();
            for (int i = 1; i < columnList.size(); i++) {
                String[] ss = columnList.get(i).split("\\.");
                String key = ss[ss.length - 1];
                keys.add(key);
            }
            dataSet = new MeasureDataSet(keys);
        }
        while (wrapper.hasNext()) {
            List<MeasureDataValue> list = new ArrayList<>();
            RowRecord record = wrapper.next();
            for (int i = 1; i < columnList.size(); i++) {
                String[] ss = columnList.get(i).split("\\.");
                String key = ss[ss.length - 1];
                if (dataSet.getMeasureKeys().contains(key)) {
                    Field field = record.getFields().get(i - 1);
                    Object val = field.getObjectValue(field.getDataType());
                    DataType dataType = DataType.valueOf(field.getDataType().name());
                    MeasureDataValue dataValue = new MeasureDataValue(record.getTimestamp(), key, val, dataType);
                    list.add(dataValue);
                }
            }
            dataSet.addDataValues(list);
        }
        return dataSet;
    }
}
